package com.gossip.manager;


import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.OutputStreamWriter;
import java.net.InetAddress;
import java.net.Socket;
import java.util.ArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.management.Notification;
import javax.management.NotificationListener;

import org.json.JSONArray;
import org.json.JSONObject;

import com.gossip.GossipMember;
import com.gossip.GossipService;
import com.gossip.GossipSettings;
import com.gossip.LocalGossipMember;
import com.gossip.RemoteGossipMember;
import com.gossip.StartupSettings;

import java.util.Date;

public class GossipManager extends Thread implements NotificationListener {
	/** The maximal number of bytes the packet with the GOSSIP may be. (Default is 100 kb) */
	public static final int MAX_PACKET_SIZE = 102400;
	
	/** The list of members which are in the gossip group (not including myself). */
	private ArrayList<GossipMember> _memberList;

	
	/** The member I am representing. */
	private LocalGossipMember _me;
	
	/** The settings for gossiping. */
	private StartupSettings _settings;
	
	/** A boolean whether the gossip service should keep running. */
	private AtomicBoolean _gossipServiceRunning;

	/** A ExecutorService used for executing the active and passive gossip threads. */
	private ExecutorService _gossipThreadExecutor;

	
	
	public GossipManager(String name, String address, int port, StartupSettings settings) {

		
		// Assign the GossipSettings to the instance variable.
		_settings = settings;
		Date date = new Date();
		// Create the local gossip member which I am representing.
		_me = new LocalGossipMember(name,address, port, date, this, 1000);
		_memberList = settings.getGossipMembers();
		
		// Print the startup member list when the service is in debug mode.
		GossipService.debug("Startup member list:");
		GossipService.debug("---------------------");
		// First print out myself.
		GossipService.debug(_me);
		
		// Set the boolean for running the gossip service to true.
		_gossipServiceRunning = new AtomicBoolean(true);
		
		// Add a shutdown hook so we can see when the service has been shutdown.
		Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
			public void run() {
				GossipService.info("Service has been shutdown...");
			}
		}));
	}
	
	@Override
	public void handleNotification(Notification notification, Object handback) {

		// Get the local gossip member associated with the notification.
		LocalGossipMember deadMember = (LocalGossipMember) notification.getUserData();

		GossipService.info("Dead member detected: " + deadMember);

		// Remove the member from the active member list.
		synchronized (this._memberList) {
			this._memberList.remove(deadMember);
		}
	}

	public StartupSettings getSettings() {
		return _settings;
	}

	public ArrayList<GossipMember> getMemberList() {
		return _memberList;
	}
	
	public LocalGossipMember getMyself() {
		return _me;
	}
	
	public void run() {
		
		GossipService.info("The GossipService is started.");

		// keep the main thread around
		while(_gossipServiceRunning.get()) {
			try {
				TimeUnit.SECONDS.sleep(15);
				sendMembershipList();
			} catch (InterruptedException e) {
				GossipService.info("The GossipClient was interrupted.");
			}
		}
	}
	
	/**
	 * Shutdown the gossip service.
	 */
	public void shutdown() {
		_gossipThreadExecutor.shutdown();
		_gossipServiceRunning.set(false);
	}
	public void sendMembershipList() 
	{
		System.out.println("before Sending MembershipList");
		sendFront();
		System.out.println("sendFront done");
		sendNext();
		System.out.println("sendNext done");
	}
	private void sendFront() {
		RemoteGossipMember frontMember = null;
		boolean changed = false;
		try {
			frontMember = getFrontMember();
			if (frontMember != null&&_memberList.size()>0) 
			{
				JSONArray arr = new JSONArray();
				ArrayList<Integer> idArr = new ArrayList<Integer>();
				for(GossipMember member: _memberList)
				{
					if(!idArr.contains(member.getMemberId()))
					{
						idArr.add(member.getMemberId());
						arr.put(member.toJSONObject());
					}
					else
					{
						GossipService.debug("warning! dup member is found in member List: " + member.toString());
					}
				}
				if(!idArr.contains(_me.getMemberId()))
				{
					idArr.add(_me.getMemberId());
					arr.put(_me.toJSONObject());
				}
				else
				{
					GossipService.error("Error! myself is found in member List: " + _me.toString());
				}
				arr.put(_me.toJSONObject());
				GossipService.info("front haertbeat is: "+ arr.toString());

				String address = frontMember.getAddress();
				String host = address.split(":")[0];
				int port = Integer.parseInt(address.split(":")[1]);

				InetAddress dest;
				dest = InetAddress.getByName(host);
				System.out.println("Sending to " + host+":"+port);
				Socket socket = new Socket(host, port);
				BufferedOutputStream bos = new BufferedOutputStream(socket.getOutputStream());
				OutputStreamWriter osw = new OutputStreamWriter(bos);
				osw.write(arr.toString());
				osw.flush();
				socket.close();
				int tmpFailedCount = frontMember.getFailedCount();
				if(tmpFailedCount>0)
				{
					frontMember.setFailedCount(0);
					changed = true;
				}
			}
			else
			{
				System.out.println("no front");
			}
		} catch (IOException e1) {
			e1.printStackTrace();
			System.out.println("Sending failed to catch!");
			if(frontMember!=null)
			{
				int failedCount = frontMember.getFailedCount()+1;
				GossipService.debug("failed to send to "+frontMember.toString()+"  failed count:" + failedCount);
				frontMember.setFailedCount(failedCount);
				frontMember.setHeartbeat();
				changed = true;
			}
		}
		if(changed)  SaveMemberList();
	}
	
	private void sendNext() {
		RemoteGossipMember nextMember = null;
        boolean changed = false; 
		try {
			nextMember = getNextMember();
			if (nextMember != null) {
				JSONArray arr = new JSONArray();
				ArrayList<Integer> idArr = new ArrayList<Integer>();
				for(GossipMember member: _memberList)
				{
					if(!idArr.contains(member.getMemberId()))
					{
						idArr.add(member.getMemberId());
						arr.put(member.toJSONObject());
					}
					else
					{
						GossipService.debug("warning! dup member is found in member List: " + member.toString());
					}
				}
				if(!idArr.contains(_me.getMemberId()))
				{
					idArr.add(_me.getMemberId());
					arr.put(_me.toJSONObject());
				}
				else
				{
					GossipService.error("Error! myself is found in member List: " + _me.toString());
				}
				GossipService.info("next haertbeat is: "+ arr.toString());
				String address = nextMember.getAddress();
				String host = address.split(":")[0];
				int port = Integer.parseInt(address.split(":")[1]);
				InetAddress dest;
				dest = InetAddress.getByName(host);
				System.out.println("Sending to " + dest);
				Socket socket = new Socket(host, port);
				BufferedOutputStream bos = new BufferedOutputStream(socket.getOutputStream());
				OutputStreamWriter osw = new OutputStreamWriter(bos);
				osw.write(arr.toString());
				osw.flush();
				socket.close();
				int tmpFailedCcount = nextMember.getFailedCount();
				if(tmpFailedCcount>0)
				{
				    nextMember.setFailedCount(0);
				    changed = true;
				}
			}
			else
			{
				System.out.println("no next");
			}

		} catch (IOException e1) {
			e1.printStackTrace();
			if(nextMember!=null)
			{
				int failedCount = nextMember.getFailedCount()+1;
				GossipService.debug("failed to send to "+nextMember.toString()+"  failed count:" + failedCount);
				nextMember.setFailedCount(failedCount);
				nextMember.setHeartbeat();
				changed = true;
			}
		}
		if(changed)  SaveMemberList();
			
	}
	
	
	private void SaveMemberList()
	{
		try {
			MemberManager.SaveMemberList(_memberList);
		} catch (IOException e) {
			GossipService.error("Failed to save member list on sendNext");
		}
	}

    private RemoteGossipMember getFrontMember()
    {
    	int myId =_me.getMemberId();
    	int min = -1;
    	RemoteGossipMember frontMember = null;
    	int maxFailedCount = _settings.getForgiveFailedCount();
    	for(GossipMember member: _memberList)
    	{
    		RemoteGossipMember rMember = (RemoteGossipMember)member;
    		if(rMember == null)
    			continue;
    		if(rMember.getFailedCount()>=maxFailedCount)
    			continue;
    		int tmpId = member.getMemberId();
    		if(tmpId>=myId) continue;
    		int tmp = myId - tmpId;
    		if(min==-1||min>tmp)
    		{
    			min = tmp;
    			frontMember = (RemoteGossipMember)member;
    		}
    	}
    	return frontMember;	
    }
    
    private RemoteGossipMember getNextMember()
    {
    	int myId =_me.getMemberId();
    	int max = -1;
    	RemoteGossipMember nextMember = null;
    	int maxFailedCount = _settings.getForgiveFailedCount();
    	for(GossipMember member: _memberList)
    	{
    		RemoteGossipMember rMember = (RemoteGossipMember)member;
    		if(rMember == null)
    			continue;
    		if(rMember.getFailedCount()>=maxFailedCount)
    			continue;
    		int tmpId = member.getMemberId();
    		if(tmpId<=myId) continue;
    		int tmp = tmpId - myId;
    		if(max==-1||max>tmp)
    		{
    			max = tmp;
    			nextMember = (RemoteGossipMember)member;
    		}
    	}
    	return nextMember;	
    }
}
